package studio.raptor.dispatcher.scheduler;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.CuratorListener;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.junit.BeforeClass;
import org.junit.Test;

/**
 * @author jack
 * @since 1.0
 */
public class CuratorListenerUtils {

  static CuratorFramework client;

  @BeforeClass
  public static void setup() {
     client = getConnection();

  }



  /**
   * 对指定的节点进行监听
   * 仅仅能监听指定的节点的数据修改，删除操作，并且只能监听一次
   *
   * @throws Exception
   */
  @Test
  public void setListenterOne() throws Exception{
    client.getData().usingWatcher(new Watcher() {
      @Override
      public void process(WatchedEvent event) {
        System.out.println("获取 two 节点 监听器 : " + event);
      }
    }).forPath("/two");
    Thread.sleep(Long.MAX_VALUE);

  }

  @Test
  public void setListenterTwo() throws Exception{
    ExecutorService pool = Executors.newCachedThreadPool();

    CuratorListener listener = new CuratorListener() {
      @Override
      public void eventReceived(CuratorFramework client, CuratorEvent event) throws Exception {
        System.out.println("监听器  : "+ event.toString());
      }
    };
    client.getCuratorListenable().addListener(listener,pool);
    client.getData().inBackground().forPath("/two");
    client.getData().inBackground().forPath("/two");
    client.getData().inBackground().forPath("/two");
    client.getData().inBackground().forPath("/two");
    Thread.sleep(Long.MAX_VALUE );

  }

  /**
   * 监视一个路径下1）孩子结点的创建、2）删除，3）以及结点数据的更新。
   *                  产生的事件会传递给注册的PathChildrenCacheListener
   * @throws Exception
   */
  @Test
  public void testPathNode() throws Exception{

    PathChildrenCache pathChildrenCache = new PathChildrenCache(client,"/server2task/192.168.199.241-9090",true);
    PathChildrenCacheListener childrenCacheListener = new PathChildrenCacheListener() {
      @Override
      public void childEvent(CuratorFramework client, PathChildrenCacheEvent event)
          throws Exception {
        ChildData data = event.getData();
        switch (event.getType()) {
          case CHILD_ADDED:
            System.out.println("child_add = [" + data.getPath() + "], data = [" + new String(data.getData() )+ "]");
            break;
          case CHILD_REMOVED:
            System.out.println("child_removed = [" + data.getPath() + "], data = [" + new String(data.getData() ) + "]");
            break;
          case CHILD_UPDATED:
            System.out.println("child_updated = [" + data.getPath() + "], data = [" + new String(data.getData() ) + "]");
            break;
          default:
            break;
        }
      }
    };

    pathChildrenCache.getListenable().addListener(childrenCacheListener);
    System.out.println("Register zk watcher successfully!");

    pathChildrenCache.start(StartMode.POST_INITIALIZED_EVENT);

    Thread.sleep(Long.MAX_VALUE );

  }

  /**
   * 监视一个结点的创建、更新、删除，并将结点的数据缓存在本地
   */
  @Test
  public void testNodeCache() throws Exception {
    final NodeCache nodeCache = new NodeCache(client,"/server2task/192.168.31.241-9090");
    NodeCacheListener nodeCacheListener = new NodeCacheListener() {
      @Override
      public void nodeChanged() throws Exception {
        System.out.println("the test node is change and result is :");
        System.out.println("path : "+nodeCache.getCurrentData().getPath());
        System.out.println("data : "+new String(nodeCache.getCurrentData().getData()));
        System.out.println("stat : "+nodeCache.getCurrentData().getStat());

      }
    };
    nodeCache.getListenable().addListener(nodeCacheListener);
    nodeCache.start();
    Thread.sleep(Long.MAX_VALUE );
  }

  @Test
  public void testTreeNode() throws Exception {
    TreeCache treeCache = new TreeCache(client,"/dispatcher/server2task/192.168.199.241-9090");
    treeCache.getListenable().addListener(new TreeCacheListener() {
      @Override
      public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
        ChildData data = event.getData();
        if(data != null) {
          switch (event.getType()) {
            case NODE_ADDED:
              System.out.println("NODE_ADDED : "+ data.getPath() +"  数据:"+ new String(data.getData()));
              break;
            case NODE_REMOVED:
              System.out.println("NODE_REMOVED : "+ data.getPath() +"  数据:"+ new String(data.getData()));
              break;
            case NODE_UPDATED:
              System.out.println("NODE_UPDATED : "+ data.getPath() +"  数据:"+ new String(data.getData()));
              break;

            default:
              break;

          }
        } else {
          System.out.println( "data is null : "+ event.getType());

        }
      }
    });

    treeCache.start();
    Thread.sleep(Long.MAX_VALUE );

  }


  private static CuratorFramework getConnection() {
    //zk 地址
    String connectString = "127.0.0.1:2181";
    // 连接时间 和重试次数
    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
    CuratorFramework client = CuratorFrameworkFactory.builder().namespace("dispatcher").retryPolicy(retryPolicy).connectString(connectString).build();
    client.start();
    return client;
  }


}
